// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/kucoin/feed_v1/processor.h"

#include <boost/algorithm/string.hpp>
#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/kucoin/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::kucoin::feed_v1::impl {

void KucoinFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("type")) {
    if (strcmp(document["type"].GetString(), "pong") == 0 ||
        strcmp(document["type"].GetString(), "welcome") == 0 ||
        strcmp(document["type"].GetString(), "ack") == 0) {
      return;
    }
    if (strcmp(document["type"].GetString(), "error") == 0) {
      LOG(INFO) << "[Kucoin Feed Processor] meta msg for " << packet.topic_name << ": "
                << packet.payload.data();
      return;
    }
  }

  if (!document.HasMember("url") && (!document.HasMember("subject") || !document.HasMember("data"))) {
    LOG(INFO) << "[Kucoin Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
    return;
  }

  upd_mgr_.onBeginPacket(packet);

  if (document.HasMember("subject")) {
    if (strcmp(document["subject"].GetString(), "level2") == 0) {  // book snapshot
      std::string native_symbol = document["topic"].GetString();
      native_symbol = native_symbol.substr(native_symbol.find(":")+1);
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (!stat.snapshot_received || !stat.update_received) {
          if (parser_config_.publish_bbo_book_only()) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
            std::optional<PriceQty> bid_post = book->Bid0();
            std::optional<PriceQty> ask_post = book->Ask0();
            bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
            if (book_generated && is_bbo_changed) {
              upd_mgr_.PublishBook();
            }
          } else if (parser_config_.has_sampling_frequency_bps()) {
            bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
            bool sampled = book->SampleByFrequency(
                book->Bid0(),
                book->Ask0(),
                parser_config_.sampling_frequency_bps());
            if (parser_config_.has_trade_sampling_frequency_bps()) {
              auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
              int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
              if (sampled ||
                  timespan > parser_config_.trade_sampling_time_length()) {
                auto* trade = upd_mgr_.GetTradeMutable();
                if (slim_trade->PopulateTradeAndClear(trade)) {
                  upd_mgr_.PublishTrade(native_symbol, false /* light bbo */);
                }
              }
            }
            if (book_generated && sampled) {
              upd_mgr_.PublishBook();
            }
          } else {
            bool book_generated = KucoinFeedUpdateParser::ParseBook(book, document["data"]);
            if (book_generated) {
              upd_mgr_.PublishBook();
            }
          }
          stat.snapshot_received = true;
        }
      }
    } else if (strcmp(document["subject"].GetString(), "trade.l2update") == 0) {  // book update
      std::string native_symbol = document["topic"].GetString();
      native_symbol = native_symbol.substr(native_symbol.find(":")+1);
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (stat.snapshot_received || packet.timestamp < SNAPSHOT_ADDED_TS) {
          if (parser_config_.publish_bbo_book_only()) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            bool book_generated = KucoinFeedUpdateParser::ParseBookL2(
                book, document["data"], true /* light bbo */);
            std::optional<PriceQty> bid_post = book->Bid0();
            std::optional<PriceQty> ask_post = book->Ask0();
            bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
            if (book_generated && is_bbo_changed) {
              upd_mgr_.PublishBook();
            }
          } else if (parser_config_.has_sampling_frequency_bps()) {
            bool book_generated = KucoinFeedUpdateParser::ParseBookL2(
                book, document["data"], true /* light bbo */);
            bool sampled = book->SampleByFrequency(
                book->Bid0(),
                book->Ask0(),
                parser_config_.sampling_frequency_bps());
            if (parser_config_.has_trade_sampling_frequency_bps()) {
              auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
              int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
              if (sampled ||
                  timespan > parser_config_.trade_sampling_time_length()) {
                auto* trade = upd_mgr_.GetTradeMutable();
                if (slim_trade->PopulateTradeAndClear(trade)) {
                  upd_mgr_.PublishTrade(native_symbol, false /* light bbo */);
                }
              }
            }
            if (book_generated && sampled) {
              upd_mgr_.PublishBook();
            }
          } else {
            bool book_generated = KucoinFeedUpdateParser::ParseBookL2(book, document["data"], false);
            if (book_generated) {
              upd_mgr_.PublishBook();
            }
          }
          stat.update_received = true;
        }
      }
    } else if (strcmp(document["subject"].GetString(), "trade.l3match") == 0) {  // trade
      std::string native_symbol = document["data"]["symbol"].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      Trade _trade{};
      int64_t timestamp = std::stoull(document["data"]["time"].GetString());
      KucoinFeedUpdateParser::ParseTrade(document["data"], &_trade, timestamp);
      if (parser_config_.has_trade_sampling_frequency_bps()) {
        auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
            packet.timestamp,
            _trade.price,
            _trade.fill_qty,
            _trade.side,
            _trade.timestamp,
            parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      } else {
        auto* trade = upd_mgr_.GetTradeMutable();
        *trade = _trade;
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      if (parser_config_.continue_on_error()) {
        LOG(ERROR) << "[Kucoin Feed Processor] unkown topic: " << packet.payload.data();
      } else {
        NOTREACHED() << "[Kucoin Feed Processor] unkown topic: " << packet.payload;
      }
    }
  } else if (document.HasMember("url")) {  // rest api
    const std::string& url = document["url"].GetString();
    // kline_rest channel
    if (url.find("/api/v1/market/candles") != std::string::npos) {
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      std::string native_symbol = parameters["symbol"];
      std::string kline_interval = parameters["type"];
      std::string channel = "?symbol={}&type=" + kline_interval;
      if (!CheckChannelFilter(channel)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      auto interval = KucoinFeedUpdateParser::ParseKlineInterval(kline_interval);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        const auto& data = document["data"]["data"].GetArray();
        int page = -1;
        if (parameters.count("currentPage") > 0) {
          page = std::stoi(parameters["currentPage"]);
        }
        if (page == 0) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        bool to_publish = false;
        for (int i = data.Size() - 1; i >= 0; i--) {
          auto& kline_data = data[i];
          bool freezed = KucoinFeedUpdateParser::ParseKlineRest(kline_data,
              klines, interval, packet.timestamp);
          to_publish = to_publish || freezed;
        }
        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    THROW() << "[Kucoin Feed Processor] unkown topic: "
            << packet.payload.data();
    return;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::kucoin::feed_v1::impl
